package webdemo;

import java.io.BufferedReader;
import java.io.FileReader;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.halo.trade.data.config.Config;
import org.halo.trade.data.sina.Level2Handler;
import org.halo.trade.data.sina.Utils;

import com.boco.base.util.datafile.CDataFile;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisCluster;

/**
 * 默认的level2的处理器
 * 由它来将数据发送至文件或者数据库
 *     目前入库的格式为：
 *          股票代码，秒或毫秒，
 *          key,数据类型,ms,数据字符串
 *     入缓存的数据为：
 *          股票代码
 *          key,数据类型,ms,数据字符串
 *          
 *     一个数据文件，分三部分，
 *     
 *     股票代码:时间戳（秒）:{
 *         行情|委托单|逐笔买|逐笔卖
 *     }
 *     
 * */
public class NewLevel2HandlerConcat implements Level2Handler{
	//数据队列
	protected ConcurrentLinkedQueue linkedQueue = new ConcurrentLinkedQueue<>();
	
	protected Jedis  jedisClusters = null ;
	//计算股票的买卖盘
	private ConcurrentHashMap<String,Integer[]> stockFile = new ConcurrentHashMap<String,Integer[]>();
	
	private CDataFile g_DataFile;
	
	public Map getStockStaticInfo(){
		Map returnMap =  new HashMap();
		Set<Entry<String, Integer[]>> setInfo = stockFile.entrySet();
		for(Entry<String, Integer[]> entry : setInfo){
			returnMap.put(entry.getKey(), entry.getValue());
		}
		return returnMap;
	}
	
	public void start(String fileSavePath){
		g_DataFile = new CDataFile();
		g_DataFile.put_FileExtensionName("DAT");
		g_DataFile.put_FullFilePath(fileSavePath);
		g_DataFile.put_FileNamePrefix("STOCK");
		g_DataFile.put_CreateFileInterval(600);
		g_DataFile.put_NumberOfMaxRecord(1000000);
		
		new DoDealMsg().start();
		jedisClusters = new Jedis(Config.REDIS_HOSTS, Config.REDIS_PORT);
		//jedisCluster = new JedisCluster(Config.getJedisCluster());
	}
	
	@Override
	public void onLevel2(int fetch_type, Map<String, Object> datas)
	{
	     return;
	}

	@Override
	public void onMessage(String msg) {
		if(msg!=null){
			//逐笔行情，是同步的所以，这里处理的有问题。
			String[] msgList = msg.split("\n");
			for (String strMsg : msgList){
				this.linkedQueue.offer(strMsg);	
			}
		}
	}
	
	/**
	 * 当数据经过一段时间没有更改时，存储至文件中。
	 * @author toriycn
	 */
	class SaveFile extends Thread{
		@Override
		public void run() {
			
		}
	}
	class DoDealMsg extends Thread{
		@Override
		public void run() {
			while(true){
				Object objMsg = linkedQueue.poll();
				//为空表示没有数据
				if (objMsg == null){
					try {
						sleep(100);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
				else{
					doMsg((String)objMsg);
				}
			}
		}
	}

	public void doMsg(String msg) {
		try {
			if (msg == null) {
				return;
			}
			if (msg.equals("")) {
				return;
			}
			//3个处理，一个是写数据文件，一个是往redis发订阅信息
			g_DataFile.writeString(msg);
			dealMsg(msg);
		} catch (Exception ex) {
			ex.printStackTrace();
			System.out.println(msg);
		}
	}

	private void dealMsg(String msg) {
		
		String[] msgKeyValue = msg.split("=");
		if (msgKeyValue.length < 2) {
			return;
		}
		// 第一位是股票代码，第二位是类型，如果没有，表示行情
		msgKeyValue[0] = msgKeyValue[0].replace("[pushmsg]", "");
		String[] stockkey = msgKeyValue[0].replace("2cn_", "").substring(2).split("_");
		// 获取时间戳
		// [2cn_sh600650=锦江投资,15:00:00,2015-10-15,38.820,37.950,42.700,37.880,42.700,PH,56293,38369848,1576583982.610,2879979,40.910,0,0.000,15176,27551087,1125549155.970,11890,16073191,666101498.440,2196,0,10,0,42.700,42.690,42.680,42.660,42.650,42.640,42.620,42.610,42.600,42.590,1378497,4800,1600,200,1600,300,200,1200,1600,500,0.000,0.000,0.000,0.000,0.000,0.000,0.000,0.000,0.000,0.000,0,0,0,0,0,0,0,0,0,0]
		// [2cn_sh600650_orders=15:00:00.000,15:00:00.000,42.700,1378497,50,0.000,0,0,43997|1700|100|22200|2200|8200|300|3000|700|500|14200|2000|500|100|500|800|100|10000|10000|800|1000|700|300|2000|1000|700|42000|100|100|600|300|1100|15000|100|2000|2200|200|2100|1000|300|5000|1000|300|100|242400|5000|1000|3000|10000|400,,,]
		// [2cn_sh600650_0=2943013|14:59:57.510|42.700|2500|106750.000|3693493|4737159|0|5]
		// [2cn_sh600650_1=2938322|14:59:49.090|42.700|50  |2135.000  |3693493|4732493|0|5]
		// [sh600650_i=A,jjtz,0.3880,0.4108,0.2090,7.0005,1557.9267,55161.0107,39056.0075,39056.0075,16105.0032,CNY,2.1421,2.2670,52.0000,1,3.6400,10.6394,1.1501]

		// _i的数据没有。
		String[] tmpMsg = msgKeyValue[1].split("[,]");
		if(tmpMsg.length<=1){
			return;
		}
		// 数据齐全了，可以写文件或和处理redis。
		String timeKey =  "";
		timeKey = tmpMsg[1];
		String[] resultMsg = new String[4];
		resultMsg[0] = stockkey[0];
		if (stockkey.length == 2) {
			resultMsg[1] = stockkey[1];
			if(resultMsg[1].equals("orders")){
				resultMsg[1] = "3";
				resultMsg[3] = msgKeyValue[1];
			}
			else{
				resultMsg[1] = stockkey[0];
				//0 和1 的数据了。
				for(String msgInfo : tmpMsg){
					String[] item = msgInfo.split("\\|");
					if (item.length < 9)
						continue;
					String resultMsgTemp ="";
					resultMsgTemp += item[0]+",";
					timeKey = item[1].substring(0,8);
					resultMsgTemp += item[1]+",";
					resultMsgTemp += item[2]+",";
					resultMsgTemp += item[3]+",";
					resultMsgTemp += item[7];
					if(stockFile.containsKey(stockkey[0])){
						Integer[] inter=stockFile.get(stockkey[0]);
						inter[Integer.parseInt(item[7])] = Integer.parseInt(item[3]);
						stockFile.put(stockkey[0], inter);
					}else{
						Integer[] inter=new Integer[3];
						inter[Integer.parseInt(item[7])] = Integer.parseInt(item[3]);
						stockFile.put(stockkey[0], inter);
					}
					resultMsg[3] += (resultMsgTemp+"|");
				}
			}
		} else {
			resultMsg[1] = "2";
			try {
				if(tmpMsg.length>=26){
					float sell = Float.parseFloat(tmpMsg[26]);
					float buy = Float.parseFloat(tmpMsg[46]);
					String sprateRate = String.format("%.4f", (buy - sell) / sell);
					msgKeyValue[1] += ("," + sprateRate);
				}
				else{
					return;
				}
			} catch (Exception ex) {
				ex.printStackTrace();
			}
			resultMsg[3] = msgKeyValue[1];
		}

		//去除时间的毫秒值
		resultMsg[2] = timeKey.substring(0, 7);
		String date = Utils.getTodayStr();
		// 对数据做一下计算，放到redis里
		String stockMsg = String.format("%s,%s,%s,%s", resultMsg[0],date,resultMsg[2], resultMsg[3]);
		jedisClusters.publish(resultMsg[0], stockMsg);
		
	};
	
	public static void main(String[] args) throws InterruptedException {
		NewLevel2HandlerConcat handler = new NewLevel2HandlerConcat();
		handler.start( "/Users/toriycn/workspace/studio/codetest/test/");
		FileReader reader;
		try {
			reader = new FileReader("/Users/toriycn/workspace/studio/javadefault/oschina/sinav2/002214.txt");
			BufferedReader br = new BufferedReader(reader);
	        String str = null;
	        while((str = br.readLine()) != null) {
	    		handler.doMsg(str);
	        }
	        br.close();
	        reader.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
        Thread.sleep(1000*1000);
	}
}
